package dht.threads;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.LinkedList;
import dht.interfaceThread.InterPut;
import dht.message.Message;
import dht.message.PutMessage;
import dht.node.Node;
import dht.tools.Duo;
import dht.tools.OperationsZone;
import dht.communication.CommunicationDHT;
import dht.communication.PortThread;
import dht.data.*;

public class ManagePut extends Thread implements InterPut{

	/**
	 * Tableau des jobs que l'on peut donner au ManagePut
	 */
	private LinkedList<Duo<TabBigInteger,DhtData<Integer>>> jobs=new LinkedList<Duo<TabBigInteger,DhtData<Integer>>>();
	
	/**
	 * Socket de communication du trhead
	 */
	private DatagramSocket maSocket;

	/**
	 * Booléen indiquant si le noeud est pret ou non
	 */
	public Boolean isBusy=new Boolean(Boolean.TRUE);
	
	
	/**
	 * le port de la socket
	 */
	public static long portSocket;

	/**
	 * Fonction de réception d'un message PUT
	 */
	public void receiveRequestPut() {
		PutMessage p=null;
		try {
			p= (PutMessage) CommunicationDHT.receiveMsg(maSocket);
			Options.update_socketPutTimeout(false, true);
			maSocket.setSoTimeout(Options.socket_put_timeout);
		} catch (SocketTimeoutException e) {
			Options.update_socketPutTimeout(true, false);
			try {
				maSocket.setSoTimeout(Options.socket_put_timeout);
			} catch (SocketException e1) {
				e1.printStackTrace();
			}
		} catch (IOException e) {
			//e.printStackTrace();
		} catch (ClassNotFoundException e) {
			//e.printStackTrace();
		}
		if (p==null) {
			return;
		}
		if (p.TYPE_MSG==Message.REPLY_Message) {
			Node.putPutReply(true);
			return;
		}
		boolean b=false;
		synchronized(Node.myContext){
			b=OperationsZone.isInZone(p.getDestination(), Node.myContext.getZoneLocale());
		}
		if (b) {
			addData(p); //méthode d'ajout d'une donnée
			ackClient(p); //méthode d'acquittement du client
		}
		else {
			int index=OperationsZone.IsInContexteNonFusion(p.getDestination());
			if (index!=-1) {
				synchronized (Node.contexteNonFusion.get(index)) {
					Node.contexteNonFusion.get(index).getDhtLocale().put(p.getDestination(), p.getDhtData());
				}
				ackClient(p); //méthode d'acquittement du client
			}
			else {
				PutMessage msg = new PutMessage(p.getadresseIpExp(),p.getVirtualId(),p.getDestination(),p.getDhtData());
				this.sendRequestPut(msg);
			}
		}
		
	}

	/**
	 * Fonction d'acquittement du client ayant fait le Put
	 * @param p le message reçu à traiter
	 */
	private void ackClient(PutMessage p) {
		InetAddress inet=null;
		try {
			inet = InetAddress.getByName(CommunicationDHT.getLocalIp());;
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}
		PutMessage msg = new PutMessage(inet,"",p.getDestination(),p.getDhtData());
		msg.TYPE_MSG=Message.REPLY_Message;
		DatagramPacket packet=CommunicationDHT.packetForgeToKnownNode(msg, p.getadresseIpExp(), PortThread.PORT_PUT);
		CommunicationDHT.sendMessage(packet, maSocket);
	}

	/**
	 * Fonction d'ajout d'une donnée à la DHT
	 * @param p le message p reçu à traiter
	 */
	private void addData(PutMessage p) {
		synchronized(Node.myContext) {
			System.out.println("Data " + p.getDhtData().toString() + " added\n");
			Node.myContext.getDhtLocale().put(p.getDestination(), p.getDhtData());
		}
	}

	
	/**
	 * Fonction appellée par le noeud pour donner un job au ManagePUT
	 * @param job le job à donner au ManagePUT
	 */
	public void putJob(Duo<TabBigInteger, DhtData<Integer>> job) {
		synchronized (jobs) {
			jobs.offer(job);
		}
		
	}

	public void run() {
		try {
			maSocket=new DatagramSocket(PortThread.PORT_PUT);
			maSocket.setSoTimeout(Options.socket_put_timeout);  // set the socket timeout to 200 ms
		} catch (SocketException e1) {
			e1.printStackTrace();
		}

		while (true) {

			boolean b=true;
			while (b) {
				synchronized(isBusy) {
					if (isBusy.booleanValue()!=true) {
						b=false;
					}
				}
				try {
					if (b!=false) sleep(10);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			Duo<TabBigInteger, DhtData<Integer>> toTreat=null;
			synchronized(jobs) {
				if (jobs.size()!=0) {
					toTreat=jobs.poll();
				}
			}
			if (toTreat!=null) {
				boolean truc=false;
				
				synchronized (Node.myContext) {
					if (OperationsZone.isInZone(toTreat.getFirst(), Node.myContext.getZoneLocale())) {
						truc=true;
					}
				}
				if (truc == true) {
					synchronized (Node.myContext) {
						Node.myContext.getDhtLocale().put(toTreat.getFirst(), toTreat.getSecond());
						Node.putPutReply(true);
					}
				}
				else {
					InetAddress monIP=null;
					try {
						monIP = InetAddress.getByName(CommunicationDHT.getLocalIp());;
					} catch (UnknownHostException e) {
						e.printStackTrace();
					}
					PutMessage msg = null;
	
					synchronized(Node.myContext){
						msg = new PutMessage(monIP,Node.myContext.getVirtualID(),toTreat.getFirst(),toTreat.getSecond());
					}
	
					this.sendRequestPut(msg);
				}
			}
			this.receiveRequestPut();
		}
	}

	
	@Override
	public boolean sendRequestPut(PutMessage p) {

		DatagramPacket packet = CommunicationDHT.packetForgeToClosestOrRandomNode(p, PortThread.PORT_PUT);
		return CommunicationDHT.sendMessage(packet, maSocket);
	}
}
